import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author : Ashiamd email: ashiamd@foxmail.com
 * @date : 2021/2/3 3:56 PM
 */
@Slf4j
public class TableTest2_1_CommonApi {
    private  final Logger logger = LoggerFactory.getLogger(TableTest2_1_CommonApi.class);
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1
        env.setParallelism(1);

        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //ExecutionEnvironment batchEnv = ExecutionEnvironment.getExecutionEnvironment();

        //流处理
        EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, blinkStreamSettings);

        //VIEW
        Table table = tableEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("id",DataTypes.INT()),
                        DataTypes.FIELD("name",DataTypes.STRING()),
                        DataTypes.FIELD("age",DataTypes.DOUBLE())
                ),
                Row.of(1, "zs", 18.2),
                Row.of(2, "bb", 28.2),
                Row.of(3, "cc", 16.2),
                Row.of(4, "dd", 38.2)
        );

        tableEnv.createTemporaryView("SourceTable", table);

        log.info("==2=================================================================");
        // Create a Table object from a SQL query
        Table table3 = tableEnv.sqlQuery("SELECT * FROM SourceTable where id > 2");
        TableResult tableResult = table3.execute();
        tableResult.collect().forEachRemaining(t -> {
            log.info("" + t.getField("id"));
        });
        //tableResult.print();
    }
}
